package com.hyt.it.ogt.kq.service.canal;

import com.alibaba.otter.canal.protocol.CanalEntry;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 处理canal监听到的表，派遣给具体的处理者
 *
 * @author zhuomiao
 * @since 2022-11-09
 */
@Component
@Slf4j
public class CanalDispatcher {

    @Autowired
    private ApplicationContext applicationContext;

    private final Map<String, CanalTableChangeHandlerService> canalTableChangeHandlerServiceMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        Map<String, CanalTableChangeHandlerService>  canalTableChangeHandlerServices = applicationContext.getBeansOfType(CanalTableChangeHandlerService.class);
        canalTableChangeHandlerServices.forEach((beanName, canalTableService) -> {
            canalTableChangeHandlerServiceMap.put(canalTableService.getHandleTableName(), canalTableService);
        });
    }

    public void dispatch(List<CanalEntry.Column> beforeColumns,
                         List<CanalEntry.Column> afterColumns,
                         String dbName, String tableName,
                         CanalEntry.EventType eventType, long timestamp) {
        CanalTableChangeHandlerService canalTableChangeHandlerService = canalTableChangeHandlerServiceMap.get(tableName);
        if (Objects.nonNull(canalTableChangeHandlerService)) {
            log.debug("Canal监测到更新:【{}】库的【{}】表 操作类型:{},操作时间{}", dbName, tableName, eventType, timestamp);
            canalTableChangeHandlerService.dataDetails(beforeColumns, afterColumns, dbName, tableName, eventType, timestamp);
        } else {
            log.debug("表{}没有处理者", tableName);
        }
    }

}
